package com.weimob.mcloud.wechatwork.customer.mq.handler.employee;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.weimob.marketing.syswechatworkusermanagement.wechatworkusermanagement.service.user.dto.req.base.EmpDepAuthorizeEnum;
import com.weimob.marketing.syswechatworkusermanagement.wechatworkusermanagement.service.user.dto.req.base.EmpDepEventTypeEnum;
import com.weimob.marketing.syswechatworkusermanagement.wechatworkusermanagement.service.user.dto.req.base.EmployeeSeatActiveEnum;
import com.weimob.mcloud.wechatwork.core.adapter.pojo.dto.channel.response.CustomerGroupListResBO;
import com.weimob.mcloud.wechatwork.core.base.MultiResponse;
import com.weimob.mcloud.wechatwork.core.entity.privatesphere.PrivateSphere;
import com.weimob.mcloud.wechatwork.core.entity.relation.employee.Employee;
import com.weimob.mcloud.wechatwork.core.entity.relation.group.CustomerGroup;
import com.weimob.mcloud.wechatwork.core.enums.employee.EmployeeActiveStatusEnum;
import com.weimob.mcloud.wechatwork.core.enums.employee.EmployeeStatusEnum;
import com.weimob.mcloud.wechatwork.core.enums.group.GroupAuthorizationStatusEnum;
import com.weimob.mcloud.wechatwork.core.service.employee.EmployeeService;
import com.weimob.mcloud.wechatwork.customer.domain.EmployeeAuthorizeChangeInfo;
import com.weimob.mcloud.wechatwork.customer.externalDependencyProxy.usermanagement.EmployeeProxy;
import com.weimob.mcloud.wechatwork.customer.mq.model.ChatGroupFacade;
import com.weimob.mcloud.wechatwork.customer.mq.model.RemoteChatGroupServiceProxy;
import com.weimob.mcloud.wechatwork.customer.service.user.constant.RedisKeyConstant;
import com.weimob.mcloud.wechatwork.customer.third.PrivateSphereThird;
import com.weimob.mcloud.wechatwork.customer.third.core.CustomerGroupExportServiceProxy;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author yu liu  2022/5/24 23:59
 * Description
 * Version 1.0.0
 **/
@Service
@Slf4j
public class EmployeeAuthorizeChangeForGroupHandler {

    @Resource
    private RemoteChatGroupServiceProxy remoteChatGroupServiceProxy;

    @Reference
    EmployeeService employeeService;

    @Resource
    private ChatGroupFacade chatGroupFacade;

    @Value("${init.group.partition.count:50}")
    private Integer partitionCount;

    @Value("${every.employee.max.group.size:500}")
    private Integer maxGroupSize;

    @Value("${group.process.seat.timeout.minutes:720}")
    private Integer timeoutMinutes;

    @Autowired
    @Qualifier("addSeatForGroupThreadPool")
    private ThreadPoolTaskExecutor addSeatForGroupThreadPool;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private EmployeeProxy employeeProxy;

    @Autowired
    private PrivateSphereThird privateSphereThird;

    @Autowired
    private CustomerGroupExportServiceProxy customerGroupExportServiceProxy;

    public void handleEmployeeAuthorizeChange(EmployeeAuthorizeChangeInfo employeeAuthorizeChangeInfo) {
        // 1、前置校验
        Long bosId = employeeAuthorizeChangeInfo.getBosId();
        List<String> employeeIds = employeeAuthorizeChangeInfo.getEmployeeIds();
        if (CollectionUtils.isEmpty(employeeIds) || Objects.isNull(bosId)) {
            log.error("客户群-处理授权员工变更失败，employeeIds或bosId为空");
            return;
        }
        // 默认不消费全量数据
        if (EmpDepEventTypeEnum.FULL_DATA_CHANGE_EVENT.getEventType().equals(employeeAuthorizeChangeInfo.getEventType())) {
            return;
        }

        // 2、查询私域信息
        PrivateSphere privateSphere = privateSphereThird.fetchByBosId(bosId);
        if (Objects.isNull(privateSphere)) {
            log.info("客户群-处理授权员工变更失败，查询私域信息失败 bosId : {}", bosId);
            return;
        }

        // 3、查询员工信息
        // cqq 已调整 employeeIds.size 永远等于 1
        MultiResponse<Employee> employeeMultiResponse = employeeProxy.getByEmployeeIdOrTxUserIds(privateSphere, employeeIds, null);
        if (Objects.isNull(employeeMultiResponse) || CollectionUtils.isEmpty(employeeMultiResponse.getDataList())) {
            log.error("客户群-处理授权员工变更，查询员工信息失败:{}", JSON.toJSONString(employeeAuthorizeChangeInfo));
            return;
        }
        List<Employee> employeeList = employeeMultiResponse.getDataList();

        // 4、具体处理逻辑
        // 处理客户群逻辑,有坐席的必然再授权范围内，在授权范围内不一定有坐席，只需处理员工增加坐席，因为授权范围变更会进行客户群数据的同步，此处无需重复同步
        // 激活坐席
        if (EmployeeSeatActiveEnum.EMPLOYEE_SEAT_ACTIVE.getChangeType().equals(employeeAuthorizeChangeInfo.getChangeType())) {
            activeSeat(employeeList, privateSphere);
            return;
        }
        // 移除授权员工/取消坐席
        if ((EmpDepAuthorizeEnum.EMPLOYEE_UNAUTHORIZE.getChangeType().equals(employeeAuthorizeChangeInfo.getChangeType())
                || EmployeeSeatActiveEnum.EMPLOYEE_SEAT_INVALIDATE.getChangeType().equals(employeeAuthorizeChangeInfo.getChangeType()))) {
            remove(employeeList, privateSphere);
        }
    }

    /**
     * 移除授权范围/取消坐席逻辑
     *
     * @param employees
     * @param privateSphere
     * @return
     */
    private boolean remove(List<Employee> employees, PrivateSphere privateSphere) {
        for (Employee employeeInfo : employees) {
            if(Objects.isNull(employeeInfo)){
                log.error("EmployeeAuthorizeChangeForGroupHandler.remove employee is null ");
                continue;
            }
            // 1、判断员工状态 当前状态为未授权/未激活时才处理
            if (!(EmployeeStatusEnum.USER_UN_AUTHORIZATION.equals(employeeInfo.getStatus()) || EmployeeActiveStatusEnum.INACTIVE.equals(employeeInfo.getActiveStatus()))) {
                log.info("EmployeeAuthorizeChangeForGroupHandler.remove 员工状态不一致 employeeId: {}",employeeInfo.getId());
                continue;
            }
            int pageNum = 0;
            int pageSize = 500;
            MultiResponse<CustomerGroup> multiResponse;
            do {
                // 1、查群主为当前员工且状态为"有效"的客户群
                pageNum++;
                multiResponse = getCustomerGroupMultiResponse(privateSphere, employeeInfo, pageNum, pageSize);
                List<CustomerGroup> customerGroupList = multiResponse.getDataList();
                if (CollectionUtils.isEmpty(customerGroupList)) {
                    break;
                }
                // 2、修改群授权状态
                for (CustomerGroup group : customerGroupList) {
                        group.setAuthorizationStatus(GroupAuthorizationStatusEnum.UNAUTHORIZED);
                        group.setPrivateSphere(privateSphere);
                }
                // 3、更新DB
                customerGroupExportServiceProxy.updateGroupAuthorizationStatus(customerGroupList);
            } while ((long) pageNum * pageSize < multiResponse.getTotalCount());
        }
        return false;
    }

    private MultiResponse<CustomerGroup> getCustomerGroupMultiResponse(PrivateSphere privateSphere, Employee employeeInfo, int pageNum, int pageSize) {
        CustomerGroup customerGroup = new CustomerGroup();
        customerGroup.setPrivateSphere(privateSphere);
        Employee employee = new Employee();
        employee.setId(employeeInfo.getId());
        customerGroup.setOwner(employee);
        customerGroup.setAuthorizationStatus(GroupAuthorizationStatusEnum.AUTHORIZED);
        HashMap<String, Object> extInfo = new HashMap<>(16);
        extInfo.put("orderBy", "id");
        extInfo.put("direction", "asc");
        return customerGroupExportServiceProxy.fetchCustomerGroupList(customerGroup, extInfo, null, null, false, true, pageNum, pageSize, false);
    }

    /**
     * 激活坐席逻辑
     *
     * @param privateSphere
     * @param employees
     * @return
     */
    private Boolean activeSeat(List<Employee> employees, PrivateSphere privateSphere) {
        // 私域判空
        if(Objects.isNull(privateSphere) || StringUtils.isBlank(privateSphere.getCorpId())){
            log.error("EmployeeAuthorizeChangeForGroupHandler.activeSeat，私域或者corpId为空！");
            return false;
        }

        // 1、过滤员工状态有效的数据
        List<Employee> validEmployees = employees.stream()
                .filter(item -> Objects.nonNull(item) && EmployeeStatusEnum.ACTIVATED.equals(item.getStatus()) && EmployeeActiveStatusEnum.ACTIVE.equals(item.getActiveStatus()))
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(validEmployees)) {
            return false;
        }

        // 2、新增坐席 12h小时内只处理一次 (注：线上出现大量重复消息)
        List<Employee> waitDealEmployees = new ArrayList<>();
        ValueOperations<String, String> valueOperations = stringRedisTemplate.opsForValue();
        for (Employee employee : validEmployees) {
            if(StringUtils.isBlank(employee.getId())){
                log.error("EmployeeAuthorizeChangeForGroupHandler.activeSeat，employee 为空");
                continue;
            }
            String redisKey = String.format(RedisKeyConstant.GROUP_PROCESS_SEAT_INTERVAL_LOCK, privateSphere.getBosId(), employee.getId());
            Boolean lockSuccess = valueOperations.setIfAbsent(redisKey, String.valueOf(System.currentTimeMillis()), timeoutMinutes, TimeUnit.MINUTES);
            if (Boolean.TRUE.equals(lockSuccess)) {
                waitDealEmployees.add(employee);
            } else {
                log.info("客户群-处理授权员工变更，12h内已经激活过坐席 employeeId: {}", employee.getId());
            }
        }
        if (CollectionUtils.isEmpty(waitDealEmployees)) {
            return false;
        }

        // 3、调腾讯-根据UserId拉群客户群&初始化客户群信息
        List<String> ownerList = waitDealEmployees.stream().map(Employee::getTxUserId).collect(Collectors.toList());
        List<CustomerGroupListResBO.GroupChat> groupChatList = remoteChatGroupServiceProxy.getAllCorpGroupChatList(privateSphere, ownerList);
        if (CollectionUtils.isEmpty(groupChatList)) {
            return false;
        }

        // 3、同步群信息
        // 当一个员工的客户群过多时（例：某个员工的客户群数量为3700），会出现超过5分钟还未处理完的情况，Artemis会reBalance 一直卡住
        // 改为当一个员工客户群过多时异步处理
        if (groupChatList.size() > maxGroupSize) {
            log.info("员工的客户群过多 corpId:{} userIdList: {} groupChatList.size: {}", privateSphere.getCorpId(), JSON.toJSONString(ownerList), groupChatList.size());
            addSeatForGroupThreadPool.execute(() -> chatGroupFacade.syncData(privateSphere, groupChatList));
        } else {
            chatGroupFacade.syncData(privateSphere, groupChatList);
        }
        return true;
    }

}
